package com.wl.mr.comment.step3;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

public class CommentRecordWrite extends RecordWriter<CommentBean, NullWritable> {

    private FSDataOutputStream goodOut;
    FSDataOutputStream commentOut;
    FSDataOutputStream badOut;

    public CommentRecordWrite(FSDataOutputStream goodOut, FSDataOutputStream commentOut, FSDataOutputStream badOut) {
        this.goodOut = goodOut;
        this.commentOut = commentOut;
        this.badOut = badOut;
    }

    @Override
    public void write(CommentBean key, NullWritable value) throws IOException, InterruptedException {
        int commentStatus = key.getCommentStatus();
        String bean = key.toString();

        if(commentStatus == 0){
            goodOut.write(bean.getBytes());
            goodOut.write("\n".getBytes());
        }else if(commentStatus == 1){
            commentOut.write(bean.getBytes());
            commentOut.write("\n".getBytes());
        }else {
            badOut.write(bean.getBytes());
            badOut.write("\n".getBytes());
        }
    }

    @Override
    public void close(TaskAttemptContext context) throws IOException, InterruptedException {
        IOUtils.closeStream(goodOut);
        IOUtils.closeStream(commentOut);
        IOUtils.closeStream(badOut);
    }
}
